package com.raven.play.stream.test.scaladsl

import akka.Done
import akka.actor.ActorSystem
import akka.stream.alpakka.slick.scaladsl._
import akka.stream.scaladsl.{Sink, Source}
import org.scalatest.time.SpanSugar.convertIntToGrainOfTime
import org.slf4j.LoggerFactory
import slick.basic.DatabaseConfig
import slick.jdbc.H2Profile.api._
import slick.jdbc.{GetResult, JdbcProfile}

import scala.concurrent.{Await, ExecutionContext, Future}
import scala.util.{Failure, Success}

/**
 * 一种重要的例子
 * 将数据写入数据库, 再读出
 *
 * 在主函数下, 一个方法中写入数据库, 再读出没有任何问题
 * 在Junit5下, 会报: 流处理失败: Database is already closed (to disable automatic closing at VM shutdown, add ";DB_CLOSE_ON_EXIT=FALSE" to the db URL) [90121-224]
 * 在Junit4下, 连错都不报, 写入数据库后直接结束
 */
object JdbcTest1 extends App {

  val logger = LoggerFactory.getLogger(this.getClass)

  // case class, 声明对象结构, 就像 java Lombok 里的 @Data 数据对象
  case class User(id: Int, name: String)

  implicit val system = ActorSystem.create("slick-example")
  implicit val ec: ExecutionContext = system.dispatcher

  // 根据配置生成 session 对象
  // implicit val session: SlickSession = SlickSession.forConfig("slick-h2")
  val databaseConfig = DatabaseConfig.forConfig[JdbcProfile]("slick-h2")
  implicit val session: SlickSession = SlickSession.forConfig(databaseConfig)

  // 注册清理事件: session 关闭
  system.registerOnTermination { () =>
    logger.info("system is terminating")
    session.close()
    logger.info("session is terminated")
    logger.info("system is terminated")
  }

  val users = (1 to 3).map(i => User(i, s"Name$i"))
  val source = Source(users)

  // This import enables the use of the Slick sql"...",
  // sqlu"...", and sqlt"..." String interpolators.
  // See "http://slick.lightbend.com/doc/3.2.1/sql.html#string-interpolation"
  //  import session.profile.api._

  // 创建表
  val createTableDone = session.db.run(sqlu"CREATE TABLE IF NOT EXISTS AKKA_TEST_USER (id INT PRIMARY KEY, name VARCHAR(255))")
  // 同步等待创建表的操作完成
  Await.result(createTableDone, 30.seconds)

  // 清理表, 测试用的, 防止主键冲突
  val truncateTableDone = session.db.run(sqlu"TRUNCATE TABLE AKKA_TEST_USER")
  // 同步等待创建表的操作完成
  Await.result(truncateTableDone, 30.seconds)

  // 创建 Sink, 接收 User 对象, 写入数据库
  val slickSink: Sink[User, Future[Done]] = Slick.sink(user =>
    sqlu"INSERT INTO AKKA_TEST_USER VALUES(${user.id}, ${user.name})"
  )

  // 组装新增数据的流程图
  val insertDone: Future[Done] = source
    .alsoTo(Sink.foreach(it => logger.info(it.toString)))
    .runWith(slickSink)

  Await.result(insertDone, 30.seconds)

  // 隐含组装对象
  //  implicit val getUserResult: GetResult[User] = GetResult(r => User(r.nextInt(), r.nextString()))
  //  implicit val getTupleResult: GetResult[(Int, String)] = GetResult(r => (r.nextInt(), r.nextString()))
  implicit val getVectorResult: GetResult[Vector[Any]] = GetResult(r => Vector(r.nextInt(), r.nextString()))

  // 组装查询数据的流程图
  val selectDone: Future[Done] =
    Slick
      //      .source(sql"SELECT ID, NAME FROM AKKA_TEST_USER".as[User])
      //      .source(sql"SELECT ID, NAME FROM AKKA_TEST_USER".as[(Int, String)])
      .source(sql"SELECT ID, NAME FROM AKKA_TEST_USER".as[Vector[Any]])
      //      .log("user") // .log 需要在 application.conf 或 logback-test.xml 设置日志级别为 DEBUG
      .runForeach(it => logger.info(it.toString))

  // 处理Future的结果
  selectDone.onComplete {
    case Success(_) =>
      logger.info("流处理成功完成")
      system.terminate() // 终止Actor系统
    case Failure(exception) =>
      logger.error(s"流处理失败: ${exception.getMessage}")
      system.terminate() // 终止Actor系统
  }
}
